package com.zhang.table;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.Json;
import org.apache.flink.table.descriptors.Kafka;
import org.apache.flink.table.descriptors.Schema;
import org.apache.flink.types.Row;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @title: 读取kafka数据
 * @author: zhang
 * @date: 2022/2/10 20:14
 */
public class KafkaSource {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);

        // todo 使用连接器方式读取kafka数据
        tableEnvironment.connect(new Kafka()
                .version("universal")
                .topic("test_flink_table")
                .startFromLatest()
                .property(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop302:9092")
                .property(ConsumerConfig.GROUP_ID_CONFIG,"zhangzhang"))
                .withSchema(new Schema()
                        .field("word", DataTypes.STRING())
                        .field("count",DataTypes.INT()))
                //.withFormat(new Csv())
                .withFormat(new Json())
                .createTemporaryTable("wordCount");

        //todo 3 使用连接器创建表
        Table table = tableEnvironment.from("wordCount");

        //todo 4 查询
        Table select = table.groupBy($("word"))
                .select($("word"), $("count").sum());


        tableEnvironment.toRetractStream(select, Row.class).print();

        env.execute();
    }
}
